package com.company.wushuo.justfortest.reactive.internal.operators;

import com.company.wushuo.justfortest.reactive.Observable;
import com.company.wushuo.justfortest.reactive.OnSubscribe;
import com.company.wushuo.justfortest.reactive.Subscriber;
import com.company.wushuo.justfortest.reactive.function.Func1;

/**
 * Created by wushuo on 2021/4/6
 */
public class OnSubscribeMap<T, R> implements OnSubscribe<R> {

    private Observable<T> source;
    private Func1<T, R> func1;

    public OnSubscribeMap(Observable<T> source, Func1<T, R> func1) {
        this.source = source;
        this.func1 = func1;
    }

    @Override
    public void call(final Subscriber<R> subscriber) {
        source.subscribe(new Subscriber<T>() {
            @Override
            public void onNext(T value) {
                try {
                    R val = func1.call(value);
                    subscriber.onNext(val);
                } catch (Exception e) {
                    subscriber.onError(e);
                    unsubscribe();
                }
            }

            @Override
            public void onError(Throwable throwable) {
                subscriber.onError(throwable);
            }

            @Override
            public void onComplete() {
                subscriber.onComplete();
            }
        });
    }
}
